Datasource: Add file_metadata() to DataSink for per-file write metadata by qzyu999 · Pull Request #23656 · apache/datafusion

qzyu999 · 2026-07-17T01:01:36Z

Which issue does this PR close?

Rationale for this change

External table formats (Apache Iceberg, Delta Lake, Apache Hudi) require per-file column statistics (column_sizes, null_counts, lower/upper bounds, split_offsets) when committing written files to their catalogs. Currently, ParquetSink computes and stores this metadata internally (self.written) but the DataSink trait only returns a row count forcing external consumers to either:

Bypass DataFusion's write pipeline entirely (use PyArrow/parquet-rs directly)
Re-read Parquet footers after writing (additional I/O round-trips)
Downcast DataSinkExec.sink() to ParquetSink and call written() (fragile, undocumented)

This PR adds a standard, non-breaking mechanism to expose per-file metadata through the DataSink trait.

What changes are included in this PR?

New type in datafusion-datasource:
- FileWriteMetadata: path + row_count + byte_size + optional format-specific metadata bytes
Trait extension (non-breaking, default implementation):
- DataSink::file_metadata() Vec<FileWriteMetadata> (default: empty)
ParquetSink implementation:
- Overrides file_metadata() using the existing self.written HashMap that already collects ParquetMetaData during writes zero additional I/O
DataSinkExec convenience:
- file_metadata() method that delegates to the inner sink, avoiding the downcast dance

The typed ParquetMetaData remains accessible via the existing ParquetSink::written() method for Rust consumers who need full column-level statistics directly. The format_metadata: Option<Bytes> field on FileWriteMetadata is reserved for future Thrift serialization when FFI consumers (e.g. Python via PyO3) need it.

Are these changes tested?

Yes 14 new tests across both crates:

datafusion-datasource (7 tests):

FileWriteMetadata equality and clone behavior
Default file_metadata() returns empty after write_all
Overridden file_metadata() returns correct entries
Idempotent behavior (calling twice returns same result)
DataSinkExec::file_metadata() delegates correctly

datafusion-datasource-parquet (7 tests):

Empty before write, populated after write
Row count consistency (write_all count == sum of per-file row_counts)
Consistent with existing ParquetSink::written() accessor
Idempotent across multiple calls
Partitioned writes produce one entry per partition
E2E test through DataSinkExec wrapper with Arc<dyn DataSink> dispatch

All existing parquet_sink_write* tests continue to pass unchanged.

Are there any user-facing changes?

New public API surface (additive only, no breaking changes):

DataSink::file_metadata() default method
FileWriteMetadata struct
DataSinkExec::file_metadata() convenience method

codecov-commenter · 2026-07-17T04:16:36Z

Codecov Report

❌ Patch coverage is 96.21622% with 14 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.86%. Comparing base (7ca6e54) to head (a66620e).
⚠️ Report is 170 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/datasource/src/sink.rs	88.13%	14 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #23656      +/-   ##
==========================================
+ Coverage   80.66%   80.86%   +0.19%     
==========================================
  Files        1086     1098      +12     
  Lines      366694   374644    +7950     
  Branches   366694   374644    +7950     
==========================================
+ Hits       295806   302967    +7161     
- Misses      53265    53632     +367     
- Partials    17623    18045     +422

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Add FileWriteMetadata struct to datafusion-datasource along with a new default method on the DataSink trait: - file_metadata(): Post-hoc accessor returning per-file path, row count, byte size, and optional format-specific metadata bytes after write_all completes. The default implementation returns an empty Vec, so existing DataSink implementations are unaffected. ParquetSink overrides file_metadata() to expose path, row_count, and byte_size for each written file using the existing self.written HashMap that already collects ParquetMetaData during writes. The typed ParquetMetaData remains accessible via ParquetSink::written() for Rust consumers who need column-level statistics directly. DataSinkExec gains a file_metadata() convenience accessor that delegates to the underlying sink. This enables external table formats (Iceberg, Delta Lake, Hudi) to obtain per-file metadata after a DataFusion write without re-reading file footers or bypassing the write pipeline entirely. Closes apache#23472

Copilot

Pull request overview

Note

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Adds a standardized way for DataSink implementations to expose per-file write metadata (e.g., file path, row count, size, and optional format-specific bytes) so external consumers (Iceberg/Delta/Hudi) can commit written files without re-reading footers or downcasting to ParquetSink.

Changes:

Introduces FileWriteMetadata and adds a default DataSink::file_metadata() accessor (returns empty by default).
Implements file_metadata() for ParquetSink using already-collected written metadata (no extra I/O).
Adds a DataSinkExec::file_metadata() convenience method and new unit tests in both crates.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.

File	Description
datafusion/datasource/src/sink.rs	Adds `FileWriteMetadata`, extends `DataSink`/`DataSinkExec` API, and adds trait-level tests.
datafusion/datasource-parquet/src/sink.rs	Implements `ParquetSink::file_metadata()` and adds parquet-specific tests validating behavior.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

qzyu999 · 2026-07-30T02:02:42Z

+    fn build_test_ctx(store_url: &ObjectStoreUrl) -> Arc<TaskContext> {
+        let tmp_dir = tempfile::TempDir::new().unwrap();
+        let local = Arc::new(
+            LocalFileSystem::new_with_prefix(&tmp_dir)
+                .expect("should create object store"),
+        );
+
+        let session = SessionConfig::default();
+        let runtime = RuntimeEnv::default();
+        runtime
+            .object_store_registry
+            .register_store(store_url.as_ref(), local);
+
+        Arc::new(
+            TaskContext::default()
+                .with_session_config(session)
+                .with_runtime(Arc::new(runtime)),
+        )
+    }


Introduced a TestFixture struct that owns the TempDir alongside the TaskContext and ParquetSink. The _tmp_dir field keeps the directory alive for the entire test lifetime:

struct TestFixture { _tmp_dir: TempDir, ctx: Arc<TaskContext>, sink: Arc<ParquetSink>, schema: SchemaRef, }

qzyu999 · 2026-07-30T02:04:22Z

+                let row_count = parquet_meta.file_metadata().num_rows() as u64;
+                let byte_size: u64 = parquet_meta
+                    .row_groups()
+                    .iter()
+                    .map(|rg| rg.compressed_size() as u64)
+                    .sum();


Replaced both casts with try_into().unwrap_or(0):

let row_count: u64 = parquet_meta .file_metadata() .num_rows() .try_into() .unwrap_or(0); let byte_size: u64 = parquet_meta .row_groups() .iter() .map(|rg| u64::try_from(rg.compressed_size()).unwrap_or(0)) .sum();

Also updated the file_metadata_consistent_with_written test assertions to use the same safe conversion pattern.

qzyu999 · 2026-07-30T02:04:47Z

+    /// Sum of compressed row group sizes in bytes.
+    ///
+    /// Note: this may differ slightly from the actual on-disk file size as it
+    /// excludes the Parquet footer, page indexes, and other metadata overhead.
+    pub byte_size: u64,


Rewrote the doc to be format-neutral:

/// Best-effort total encoded data size in bytes. /// /// How this value is computed depends on the format. For Parquet it is the /// sum of compressed row-group sizes (which excludes the footer and page /// indexes); for other formats it may be the full file size or unavailable /// (zero). Consumers should treat this as an approximation. pub byte_size: u64,

timsaucer

This looks like a good addition. I have a couple of suggestions and a question. Also I think the copilot suggestions look like they should be addressed.

timsaucer · 2026-07-28T19:18:37Z

+    fn make_test_batch(schema: &SchemaRef) -> RecordBatch {
+        let col_a: ArrayRef = Arc::new(StringArray::from(vec!["foo", "bar", "baz"]));
+        let col_b: ArrayRef = Arc::new(StringArray::from(vec!["one", "two", "three"]));
+        RecordBatch::try_new(Arc::clone(schema), vec![col_a, col_b]).unwrap()


There is a record_batch! macro that makes this a little more pleasant.

Replaced manual StringArray + RecordBatch::try_new with the macro:

fn make_test_batch(_schema: &SchemaRef) -> RecordBatch { record_batch!( ("a", Utf8, ["foo", "bar", "baz"]), ("b", Utf8, ["one", "two", "three"]) ) .unwrap() }

Same for the partitioned test:

let batch = record_batch!( ("a", Utf8, ["x", "y", "x"]), ("b", Utf8, ["one", "two", "three"]) ) .unwrap();

timsaucer · 2026-07-28T19:19:43Z

+        let batch =
+            RecordBatch::try_new(Arc::clone(&schema), vec![col_a, col_b]).unwrap();


Same comment as above about reusing the record_batch! macro

timsaucer · 2026-07-28T19:21:50Z

+    /// Sum of compressed row group sizes in bytes.
+    ///
+    /// Note: this may differ slightly from the actual on-disk file size as it
+    /// excludes the Parquet footer, page indexes, and other metadata overhead.


FileWriteMetada should be format agnostic, right? Here we're talking about row group sizes, which isn't a concept in all of the datasources. I think we just need a comment update here, but if this is specific to parquet then it probably needs to go into the format_metadata.

Rewrote the doc to be format-neutral:

/// Best-effort total encoded data size in bytes. /// /// How this value is computed depends on the format. For Parquet it is the /// sum of compressed row-group sizes (which excludes the footer and page /// indexes); for other formats it may be the full file size or unavailable /// (zero). Consumers should treat this as an approximation. pub byte_size: u64,

…asts, format-agnostic docs, record_batch! macro

qzyu999 · 2026-07-30T02:09:47Z

Hi @timsaucer, thanks for all the feedback, I've responded to each of the points along with Copilot, PTAL.

timsaucer

Generally this looks good and I really like how you've expanded test coverage. In addition to my one comment (you can ignore the nit if you disagree) it looks like there are some CI failures.

timsaucer · 2026-07-30T14:30:32Z

+                    byte_size,
+                    // Full Parquet metadata is accessible via ParquetSink::written()
+                    // for Rust consumers. Thrift serialization for FFI consumers can
+                    // be added when a concrete use case (e.g. Python via PyO3) needs it.
+                    format_metadata: None,


Since this is being left for future work instead of exposing the metadata here, can you open an issue to track this and link the comment to the issue?

Done, opened #24010 to track this. Updated the code comment to link to the issue as well.

timsaucer · 2026-07-30T14:32:55Z

 }
+
+#[cfg(test)]
+mod sink_tests {


Suggested change

mod sink_tests {

mod tests {

Nit, but consistent with the repo and rust standard practice.

Good catch, fixed. Renamed to tests for consistency with the rest of the repo.

…acking issue apache#24010

github-actions Bot added the datasource Changes to the datasource crate label Jul 17, 2026

qzyu999 mentioned this pull request Jul 17, 2026

Expose per-file write metadata from DataFrame.write_parquet() apache/datafusion-python#1637

Open

qzyu999 force-pushed the datasink-write-metadata branch from ed75a23 to a4d415e Compare July 17, 2026 02:24

qzyu999 force-pushed the datasink-write-metadata branch from a4d415e to f9c560a Compare July 17, 2026 09:43

timsaucer requested review from Copilot and timsaucer July 28, 2026 18:34

Copilot AI reviewed Jul 28, 2026

View reviewed changes

Copilot started reviewing on behalf of timsaucer July 28, 2026 19:24 View session

timsaucer reviewed Jul 28, 2026

View reviewed changes

Address review feedback: fix TempDir lifetime, portable paths, safe c…

7ef1553

…asts, format-agnostic docs, record_batch! macro

timsaucer reviewed Jul 30, 2026

View reviewed changes

qzyu999 mentioned this pull request Jul 30, 2026

Populate FileWriteMetadata.format_metadata with Thrift-serialized Parquet footer for FFI consumers #24010

Open

Fix cargo fmt, rename sink_tests to tests, update Cargo.lock, link tr…

a66620e

…acking issue apache#24010

		let batch =
		RecordBatch::try_new(Arc::clone(&schema), vec![col_a, col_b]).unwrap();

Uh oh!

Conversation

qzyu999 commented Jul 17, 2026

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

codecov-commenter commented Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

timsaucer left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

qzyu999 commented Jul 30, 2026

Uh oh!

timsaucer left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

codecov-commenter commented Jul 17, 2026 •

edited

Loading